package com.bbx.flink.table_api_or_sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 连接到文件系统
 * The file system connector allows for reading and writing from
 * a local or distributed filesystem. A filesystem can be defined as:
 */
public class ConnectorTableFromFileDemo {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String createInputSql = "CREATE TABLE inputTable (\n" +
                " id String,name String,tel String " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\33.csv', \n" +
                "  'format.type' = 'csv' \n"+
                ")";


        String createOutputSql = "CREATE TABLE outputTable (\n" +
                " id String,name String,tel String " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\44.csv', \n" +
                "  'format.type' = 'csv' \n"+
                ")";

        tableEnv.executeSql(createInputSql);
        tableEnv.executeSql(createOutputSql);


        //从inputTable 读取数据,写入  outputTable
        String insertSql = "insert into outputTable (select * from  inputTable ) ";
        tableEnv.executeSql(insertSql);



        //从outputTable 中读取数据
        String querySql = "select * from  outputTable where id >30000";
        tableEnv.executeSql(querySql).print();

    }

}
